package com.garlicts.framework.binlog.sync.canal;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.Message;
import com.garlicts.framework.binlog.sync.event.DeleteCanalEvent;
import com.garlicts.framework.binlog.sync.event.InsertCanalEvent;
import com.garlicts.framework.binlog.sync.event.UpdateCanalEvent;

@Component
public class CanalSyncHandler {

	private static final Logger logger = LoggerFactory.getLogger(CanalSyncHandler.class);
	
	@Autowired
	CanalClient canalClient;
    @Value("${canal.subscribe}")
    String subscribe;
	
	@Async
	public void execute(ApplicationContext applicationContext) {
		
		CanalConnector canalConnector = canalClient.getCanalConnector();
		// subscribe格式："db.table1,db.table2"
		canalConnector.subscribe(subscribe);
		// 回滚寻找上次中断的位置
		canalConnector.rollback();
		
		while(true) {
			
			Message message = canalConnector.getWithoutAck(10);
			long batchId = message.getId();
			
			List<Entry> entries = message.getEntries();
			if(batchId == -1 || CollectionUtils.isEmpty(entries)) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                	logger.error("", e);
                }
			}else {
				
				try {
					
					entries.forEach(entry -> {
						
						if(entry.getEntryType() == EntryType.ROWDATA) {
							publishCanalEvent(applicationContext, entry);
						}
						
					});
					
					canalConnector.ack(batchId);
				
				} catch (Exception e) {
					logger.error("发送监听事件失败，回滚batchId，batchId=" + batchId, e);
					canalConnector.rollback(batchId);
					throw e;
				}				
				
			}
			
		}
	
		
	}
	
    private void publishCanalEvent(ApplicationContext applicationContext, Entry entry) {
    	
        EventType eventType = entry.getHeader().getEventType();
        switch (eventType) {
            case INSERT:
                applicationContext.publishEvent(new InsertCanalEvent(entry));
                break;
            case UPDATE:
                applicationContext.publishEvent(new UpdateCanalEvent(entry));
                break;
            case DELETE:
                applicationContext.publishEvent(new DeleteCanalEvent(entry));
                break;
            default:
                break;
        }
        
    }

}
